package org.codehaus.activemq;

import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
import javax.jms.JMSException;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TopicSession;
import javax.jms.TransactionInProgressException;
import javax.jms.XAQueueSession;
import javax.jms.XASession;
import javax.jms.XATopicSession;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.message.ActiveMQXid;
import org.codehaus.activemq.message.IntResponseReceipt;
import org.codehaus.activemq.message.ResponseReceipt;
import org.codehaus.activemq.message.XATransactionInfo;
import org.codehaus.activemq.util.IdGenerator;

public class ActiveMQXASession extends ActiveMQSession
		implements XASession, XAQueueSession, XATopicSession, XAResource {
	private static final Log log = LogFactory.getLog(ActiveMQXASession.class);
	private Xid associatedXid;
	private ActiveMQXid activeXid;

	public ActiveMQXASession(ActiveMQXAConnection theConnection) throws JMSException {
		super(theConnection, 0);
	}

	public boolean getTransacted() throws JMSException {
		return true;
	}

	public void rollback() throws JMSException {
		throw new TransactionInProgressException("Cannot rollback() inside an XASession");
	}

	public void commit() throws JMSException {
		throw new TransactionInProgressException("Cannot commit() inside an XASession");
	}

	public Session getSession() throws JMSException {
		return this;
	}

	public XAResource getXAResource() {
		return this;
	}

	public QueueSession getQueueSession() throws JMSException {
		return this;
	}

	public TopicSession getTopicSession() throws JMSException {
		return this;
	}

	public void start(Xid xid, int flags) throws XAException {
		checkClosedXA();

		if (this.associatedXid != null) {
			throw new XAException(-6);
		}

		if (((flags & 0x200000) != 2097152) || ((flags & 0x200000) == 134217728))
			;
		setXid(xid);

		XATransactionInfo info = new XATransactionInfo();
		info.setId(this.packetIdGenerator.generateId());
		info.setXid(this.activeXid);
		info.setType(101);
		try {
			this.connection.syncSendPacket(info);
		} catch (JMSException e) {
			throw toXAException(e);
		}
	}

	public void end(Xid xid, int flags) throws XAException {
		checkClosedXA();

		if ((flags & 0x2000000) == 33554432) {
			if ((this.associatedXid == null) || (this.associatedXid != xid)) {
				throw new XAException(-6);
			}

			setXid(null);
		} else if ((flags & 0x20000000) == 536870912) {
			setXid(null);
		} else if ((flags & 0x4000000) == 67108864) {
			if (this.associatedXid == xid)
				setXid(null);
		} else {
			throw new XAException(-5);
		}
	}

	public int prepare(Xid xid) throws XAException {
		if (xid == this.associatedXid) {
			throw new XAException(-6);
		}

		ActiveMQXid x = new ActiveMQXid(xid);

		XATransactionInfo info = new XATransactionInfo();
		info.setId(this.packetIdGenerator.generateId());
		info.setXid(x);
		info.setType(102);
		try {
			IntResponseReceipt receipt = (IntResponseReceipt) this.connection.syncSendRequest(info);
			return receipt.getResult();
		} catch (JMSException e) {		
			throw toXAException(e);
		}
	}

	public void rollback(Xid xid) throws XAException {
		ActiveMQXid x;
		if (xid == this.associatedXid) {
			x = this.activeXid;
		} else
			x = new ActiveMQXid(xid);

		XATransactionInfo info = new XATransactionInfo();
		info.setId(this.packetIdGenerator.generateId());
		info.setXid(x);
		info.setType(105);
		try {
			this.connection.syncSendPacket(info);
		} catch (JMSException e) {
			throw toXAException(e);
		}
	}

	public void commit(Xid xid, boolean onePhase) throws XAException {
		checkClosedXA();

		if (xid == this.associatedXid) {
			throw new XAException(-6);
		}
		ActiveMQXid x = new ActiveMQXid(xid);

		XATransactionInfo info = new XATransactionInfo();
		info.setId(this.packetIdGenerator.generateId());
		info.setXid(x);
		info.setType(onePhase ? 109 : 103);
		try {
			this.connection.syncSendPacket(info);
		} catch (JMSException e) {
			throw toXAException(e);
		}
	}

	public void forget(Xid xid) throws XAException {
		checkClosedXA();
		ActiveMQXid x;

		if (xid == this.associatedXid) {
			x = this.activeXid;
		} else
			x = new ActiveMQXid(xid);

		XATransactionInfo info = new XATransactionInfo();
		info.setId(this.packetIdGenerator.generateId());
		info.setXid(x);
		info.setType(107);
		try {
			this.connection.syncSendPacket(info);
		} catch (JMSException e) {
			throw toXAException(e);
		}
	}

	private String getResourceManagerId() {
		return ((ActiveMQXAConnection) this.connection).getResourceManagerId();
	}

	public boolean isSameRM(XAResource xaResource) throws XAException {
		if (xaResource == null) {
			return false;
		}
		if (!(xaResource instanceof ActiveMQXASession)) {
			return false;
		}
		ActiveMQXASession xar = (ActiveMQXASession) xaResource;
		return getResourceManagerId().equals(xar.getResourceManagerId());
	}

	public Xid[] recover(int flag) throws XAException {
		checkClosedXA();

		XATransactionInfo info = new XATransactionInfo();
		info.setId(this.packetIdGenerator.generateId());
		info.setType(110);
		try {
			ResponseReceipt receipt = (ResponseReceipt) this.connection.syncSendRequest(info);
			return (ActiveMQXid[]) receipt.getResult();
		} catch (JMSException e) {
			throw toXAException(e);
		}
		
	}

	public int getTransactionTimeout() throws XAException {
		checkClosedXA();

		XATransactionInfo info = new XATransactionInfo();
		info.setId(this.packetIdGenerator.generateId());
		info.setType(111);
		try {
			IntResponseReceipt receipt = (IntResponseReceipt) this.connection.syncSendRequest(info);
			return receipt.getResult();
		} catch (JMSException e) {
			throw toXAException(e);
		}
		
	}

	public boolean setTransactionTimeout(int seconds) throws XAException {
		checkClosedXA();

		XATransactionInfo info = new XATransactionInfo();
		info.setId(this.packetIdGenerator.generateId());
		info.setType(112);
		info.setTransactionTimeout(seconds);
		try {
			this.connection.asyncSendPacket(info);
			return true;
		} catch (JMSException e) {
			throw toXAException(e);
		}
		
	}

	protected void checkClosedXA() throws XAException {
		if (this.closed.get())
			throw new XAException(-7);
	}

	protected boolean isXaTransacted() {
		return true;
	}

	protected String getNextTransactionId() {
		return this.currentTransactionId;
	}

	protected void doStartTransaction() throws JMSException {
		if (this.associatedXid == null)
			throw new JMSException("Session's XAResource has not been enlisted in a distributed transaction.");
	}

	private void setXid(Xid xid) {
		if (xid != null) {
			this.associatedXid = xid;
			this.activeXid = new ActiveMQXid(xid);
			this.currentTransactionId = this.activeXid.toLocalTransactionId();
		} else {
			this.associatedXid = null;
			this.activeXid = null;
			this.currentTransactionId = null;
		}
	}

	private XAException toXAException(JMSException e) {
		if ((e.getCause() != null) && ((e.getCause() instanceof XAException))) {
			XAException original = (XAException) e.getCause();
			XAException xae = new XAException(original.getMessage());
			xae.errorCode = original.errorCode;
			xae.initCause(original);
			return xae;
		}

		XAException xae = new XAException(e.getMessage());
		xae.errorCode = -7;
		xae.initCause(e);
		return xae;
	}
}